package org.zhq.core;

import org.zhq.box.*;
import org.zhq.impl.SocketChannelAdapter;
import org.zhq.impl.async.AsyncReceiveDispatcher;
import org.zhq.impl.async.AsyncSendDispatcher;
import org.zhq.impl.bridge.BridgeSocketDispatcher;
import org.zhq.util.CloseUtils;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.net.StandardSocketOptions;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * 一个连接(可以是客户端也可以是服务器)
 *
 * @author zhengquan
 * @date 2019/1/24
 */
public abstract class Connector implements Closeable, SocketChannelAdapter.OnChannelStatusChangedListener {

    //唯一标示
    protected UUID key = UUID.randomUUID();
    //socket通道
    private SocketChannel channel;
    //发送者
    private Sender sender;
    //接收者
    private Receiver receiver;
    //发送调度者
    private SendDispatcher sendDispatcher;
    //接收调度者
    private ReceiveDispatcher receiveDispatcher;

    private final List<ScheduleTask> scheduleTaskList = new ArrayList<>(4);

    /**
     * 接收包回调方法
     */
    private ReceiveDispatcher.ReceivePacketCallback receivePacketCallback = new ReceiveDispatcher.ReceivePacketCallback() {
        /**
         *
         * 接收提供一个新包
         * @param type 包的类型
         * @param length 包的长度大小
         * @return
         */
        @Override
        public ReceivePacket<?, ?> onArrivedNewPacket(byte type, long length, byte[] headerInfo) {
            switch (type) {
                case Packet.TYPE_MEMORY_BYTES:
                    return new BytesReceivePacket(length);
                case Packet.TYPE_MEMORY_STRING:
                    return new StringReceivePacket(length);
                case Packet.TYPE_STREAM_FILE:
                    return new FileReceivePacket(length, createNewReceiveFile(length, headerInfo));
                case Packet.TYPE_STREAM_DIRECT:
                    return new StreamDirectReceivePacket(length, createNewReceiveDirectOutputStream(length, headerInfo));
                default:
                    throw new UnsupportedOperationException("不支持的传输类型:" + type);
            }
        }

        /**
         * 接收包完成回调
         * @param packet 包信息
         */
        @Override
        public void onReceivedPacketCompleted(ReceivePacket packet) {
            onReceivePacket(packet);
        }

        @Override
        public void onReceivedHeartbeat() {
            System.out.println(key.toString() + ":[Heartbeat]");
        }
    };


    /**
     * 为连接安装一个 socketChannel
     *
     * @param socketChannel
     * @throws IOException
     */
    public void setup(SocketChannel socketChannel) throws IOException {
        channel = socketChannel;

        socketChannel.configureBlocking(false);
        socketChannel.socket().setSoTimeout(1000);
        socketChannel.socket().setPerformancePreferences(1, 3, 3);
        socketChannel.setOption(StandardSocketOptions.SO_SNDBUF, 16 * 1024);
        socketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 16 * 1024);
        socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
        socketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);

        IoProvider provider = IoContext.get().getIoProvider();
        SocketChannelAdapter adapter = new SocketChannelAdapter(channel, provider, this);
        sender = adapter;
        receiver = adapter;


        sendDispatcher = new AsyncSendDispatcher(sender);
        receiveDispatcher = new AsyncReceiveDispatcher(receiver, receivePacketCallback);

        //启动接收
        receiveDispatcher.start();
    }

    public void send(String msg) {
        SendPacket packet = new StringSendPacket(msg);
        sendDispatcher.send(packet);
    }

    public void send(SendPacket packet) {
        sendDispatcher.send(packet);
    }

    public void schedule(ScheduleTask task) {
        synchronized (scheduleTaskList) {
            if (scheduleTaskList.contains(task)) {
                return;
            }
            Scheduler scheduler = IoContext.get().getScheduler();
            task.schedule(scheduler);
            scheduleTaskList.add(task);
        }
    }

    public void fireIdleTimeoutEvent() {
        sendDispatcher.sendHeartbeat();
    }

    public void fireExceptionCaught(Throwable throwable) {

    }

    public long getLastActiveTime() {
        return Math.max(sender.getLastWriteTime(), receiver.getLastReadTime());
    }

    protected void onReceivePacket(ReceivePacket packet) {
//        System.out.println(key.toString() + ":" + packet.toString());
    }

    protected abstract File createNewReceiveFile(long length, byte[] headerInfo);

    protected abstract OutputStream createNewReceiveDirectOutputStream(long length, byte[] headerInfo);


    @Override
    public void close() throws IOException {
        receiveDispatcher.close();
        sendDispatcher.close();
        sender.close();
        receiver.close();
        channel.close();
    }

    @Override
    public void onChannelClosed(SocketChannel channel) throws IOException {
        CloseUtils.close(this);
        synchronized (scheduleTaskList) {
            for (ScheduleTask scheduleTask : scheduleTaskList) {
                scheduleTask.unSchedule();
            }
            scheduleTaskList.clear();
        }
    }


    public UUID getKey() {
        return key;
    }

    public void changeToBridge() {
        if (receiveDispatcher instanceof BridgeSocketDispatcher) {
            return;
        }
        receiveDispatcher.stop();
        BridgeSocketDispatcher dispatcher = new BridgeSocketDispatcher(receiver);
        receiveDispatcher = dispatcher;
        dispatcher.start();
    }

    public void bindToBridge(Sender sender) {
        if (sender == this.sender) {
            throw new UnsupportedOperationException("不能设置当前连接的发送者");
        }
        if (!(receiveDispatcher instanceof BridgeSocketDispatcher)) {
            throw new IllegalStateException("receiveDispatcher 不是桥接调度");
        }
        ((BridgeSocketDispatcher) receiveDispatcher).bindSender(sender);
    }

    public void unBindToBridge() {
        if (!(receiveDispatcher instanceof BridgeSocketDispatcher)) {
            throw new IllegalStateException("receiveDispatcher 不是桥接调度");
        }
        ((BridgeSocketDispatcher) receiveDispatcher).bindSender(null);
    }

    public Sender getSender() {
        return sender;
    }
}
